package cn.itcast.tags.models.statistics

import cn.itcast.tags.models.{AbstractModel, ModelType}
import cn.itcast.tags.tools.TagTools
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.row_number

class PayTypeModel2 extends AbstractModel("PayTypeModel2",ModelType.STATISTICS){
  override def doTag(businessDF: DataFrame, tagDF: DataFrame): DataFrame = {
    import businessDF.sparkSession.implicits._

    /**
     * root
     * |-- memberid: string (nullable = true)
     * |-- paymentcode: string (nullable = true)
     * +--------+-----------+
     * |memberid|paymentcode|
     * +--------+-----------+
     * |13      |alipay     |
     */
    businessDF.printSchema()
    businessDF.show(10,false)
    val payDF = businessDF.groupBy(
      $"memberid", $"paymentcode"
    )
      .count()
        .withColumn(
          "rank",
          row_number().over(
            Window.partitionBy($"memberid").orderBy($"count".desc)
          )
        )
        .where($"rank"===1)
      .select($"memberid".as("id"), $"paymentcode")

    payDF.printSchema()
    payDF.show(10,false)

    val modelDF: DataFrame = TagTools.ruleMatchTag(
      payDF, "paymentcode", tagDF
    )
    modelDF
  }
}
object PayTypeModel2{
  def main(args: Array[String]): Unit = {
    val payTypeModel = new PayTypeModel2()
    payTypeModel.executeModel(350L)

  }
}